package com.cwh.taskcenter.task.consumer;

import com.alibaba.fastjson2.JSON;
import com.cwh.taskcenter.domain.notice.factory.NotificationStrategyFactory;
import com.cwh.taskcenter.domain.notice.gateway.TaskNotificationGateWay;
import com.cwh.taskcenter.domain.notice.model.SendTaskNotification;
import com.cwh.taskcenter.domain.notice.model.TaskNotification;
import com.cwh.taskcenter.domain.notice.strategy.NotificationStrategy;
import com.cwh.taskcenter.domain.user.gateway.UserGateWay;
import com.cwh.taskcenter.domain.user.model.User;
import com.cwh.taskcenter.notice.dto.ChannelConfigDTO;
import com.cwh.taskcenter.task.dto.event.TaskCompleteEvent;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author cwh
 */
@Slf4j
@Component
@AllArgsConstructor
public class TaskCompleteConsumer {

    @Autowired
    private TaskNotificationGateWay taskNotificationGateWay;

    @Autowired
    private UserGateWay userGateWay;

    // 创建线程池用于异步处理通知
    private final ExecutorService notificationExecutor = Executors.newFixedThreadPool(4);

    @Autowired
    private NotificationStrategyFactory strategyFactory;

    @KafkaListener(topics = "TASK_COMPLETE",
            groupId = "${spring.kafka.consumer.group-id}",
            concurrency = "4")
    public void consumeTaskComplete(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            TaskCompleteEvent event = parseEvent(record);
            if (event == null) {
                acknowledgment.acknowledge(); // 无效事件直接ack
                return;
            }
            User user = userGateWay.findById(event.getUserId());
            if (user == null) {
                log.warn("用户不存在, userId: {}", event.getUserId());
                return;
            }
            event.setPhone(user.getPhone());
            event.setEmail(user.getEmail());
            log.info("获取任务完成事件:{}", JSON.toJSONString(event));
            // 异步处理通知发送
            notificationExecutor.submit(() -> processNotification(event));

            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("处理任务完成事件时发生错误", e);
        }
    }

    private TaskCompleteEvent parseEvent(ConsumerRecord<String, String> record) {
        try {
            return JSON.parseObject(record.value(), TaskCompleteEvent.class);
        } catch (Exception e) {
            log.error("解析任务完成事件失败，事件内容: {}", record.value(), e);
            return null;
        }
    }

    private void processNotification(TaskCompleteEvent event) {
        try {
            TaskNotification taskNotification = taskNotificationGateWay.findByTaskId(event.getTaskId());
            if (!isNotificationValid(taskNotification)) {
                log.debug("任务通知配置无效或未启用完成通知, taskId: {}", event.getTaskId());
                return;
            }

            List<ChannelConfigDTO> channelConfigs = taskNotification.getChannelConfigs();
            if (CollectionUtils.isEmpty(channelConfigs)) {
                log.debug("任务通知渠道未配置, taskId: {}", event.getTaskId());
                return;
            }

            // 并行处理各渠道通知
            processChannels(event,channelConfigs);
        } catch (Exception e) {
            log.error("处理任务通知时发生错误, taskId: {}", event.getTaskId(), e);
        }
    }

    private boolean isNotificationValid(TaskNotification notification) {
        return notification != null && notification.isCompleteSend();
    }


    private void processChannels(TaskCompleteEvent event,List<ChannelConfigDTO> channelConfigs) {
        channelConfigs.forEach(channelConfig -> {
            SendTaskNotification sendTaskNotification = buildNotification(event, channelConfig);
            sendNotification(sendTaskNotification);
        });


    }

    private SendTaskNotification buildNotification(TaskCompleteEvent event,ChannelConfigDTO channelConfig) {
        // 获取用户邮箱
        SendTaskNotification notification = new SendTaskNotification();
        notification.setTaskId(event.getTaskId());
        notification.setUserId(event.getUserId());
        notification.setTaskName(event.getTaskName());
        notification.setChannelConfig(channelConfig);
        notification.setAccountId(1L);
        notification.setNickname("任务中心");
        notification.setMail(event.getEmail());
        notification.setPhone(event.getPhone());
        return notification;
    }

    private void sendNotification(SendTaskNotification notification) {
        ChannelConfigDTO channelConfig = notification.getChannelConfig();
        String channelType = channelConfig.getChannelType();
        try {
            NotificationStrategy strategy = strategyFactory.getStrategy(channelType);
            strategy.send(notification);
        } catch (Exception e) {
            log.error("发送{}通知失败, taskId: {}", channelType, notification.getTaskId(), e);
        }
    }

}